{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "a63a4585-8a6b-4446-9b63-8c5d5d0b80fc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import json\n",
    "\n",
    "from kafka import KafkaProducer\n",
    "\n",
    "def json_serializer(data):\n",
    "    return json.dumps(data).encode('utf-8')\n",
    "\n",
    "server = 'localhost:9092'\n",
    "\n",
    "producer = KafkaProducer(\n",
    "    bootstrap_servers=[server],\n",
    "    value_serializer=json_serializer\n",
    ")\n",
    "\n",
    "producer.bootstrap_connected()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "78bd28f9-66cb-4532-bf03-bb3fe90655b5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "--2025-03-07 19:27:06--  https://github.com/DataTalksClub/nyc-tlc-data/releases/download/green/green_tripdata_2019-10.csv.gz\n",
      "Resolving github.com (github.com)... 140.82.121.3\n",
      "Connecting to github.com (github.com)|140.82.121.3|:443... connected.\n",
      "HTTP request sent, awaiting response... 302 Found\n",
      "Location: https://objects.githubusercontent.com/github-production-release-asset-2e65be/513814948/ea580e9e-555c-4bd0-ae73-43051d8e7c0b?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=releaseassetproduction%2F20250307%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20250307T182706Z&X-Amz-Expires=300&X-Amz-Signature=6b8f2f603fe86515be24510f3f30bcf93c932b551769e5121fb0cbdf58e9b767&X-Amz-SignedHeaders=host&response-content-disposition=attachment%3B%20filename%3Dgreen_tripdata_2019-10.csv.gz&response-content-type=application%2Foctet-stream [following]\n",
      "--2025-03-07 19:27:07--  https://objects.githubusercontent.com/github-production-release-asset-2e65be/513814948/ea580e9e-555c-4bd0-ae73-43051d8e7c0b?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=releaseassetproduction%2F20250307%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20250307T182706Z&X-Amz-Expires=300&X-Amz-Signature=6b8f2f603fe86515be24510f3f30bcf93c932b551769e5121fb0cbdf58e9b767&X-Amz-SignedHeaders=host&response-content-disposition=attachment%3B%20filename%3Dgreen_tripdata_2019-10.csv.gz&response-content-type=application%2Foctet-stream\n",
      "Resolving objects.githubusercontent.com (objects.githubusercontent.com)... 185.199.108.133, 185.199.111.133, 185.199.110.133, ...\n",
      "Connecting to objects.githubusercontent.com (objects.githubusercontent.com)|185.199.108.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 8262584 (7.9M) [application/octet-stream]\n",
      "Saving to: 'green_tripdata_2019-10.csv.gz'\n",
      "\n",
      "     0K .......... .......... .......... .......... ..........  0% 1.08M 7s\n",
      "    50K .......... .......... .......... .......... ..........  1% 2.93M 5s\n",
      "   100K .......... .......... .......... .......... ..........  1% 3.15M 4s\n",
      "   150K .......... .......... .......... .......... ..........  2% 6.40M 3s\n",
      "   200K .......... .......... .......... .......... ..........  3% 5.41M 3s\n",
      "   250K .......... .......... .......... .......... ..........  3% 7.09M 3s\n",
      "   300K .......... .......... .......... .......... ..........  4% 4.84M 2s\n",
      "   350K .......... .......... .......... .......... ..........  4% 7.74M 2s\n",
      "   400K .......... .......... .......... .......... ..........  5% 20.4M 2s\n",
      "   450K .......... .......... .......... .......... ..........  6% 10.9M 2s\n",
      "   500K .......... .......... .......... .......... ..........  6% 5.03M 2s\n",
      "   550K .......... .......... .......... .......... ..........  7%  139M 2s\n",
      "   600K .......... .......... .......... .......... ..........  8% 11.8M 2s\n",
      "   650K .......... .......... .......... .......... ..........  8%  333M 1s\n",
      "   700K .......... .......... .......... .......... ..........  9% 6.83M 1s\n",
      "   750K .......... .......... .......... .......... ..........  9% 14.7M 1s\n",
      "   800K .......... .......... .......... .......... .......... 10% 4.41M 1s\n",
      "   850K .......... .......... .......... .......... .......... 11% 6.43M 1s\n",
      "   900K .......... .......... .......... .......... .......... 11%  292M 1s\n",
      "   950K .......... .......... .......... .......... .......... 12% 2.94M 1s\n",
      "  1000K .......... .......... .......... .......... .......... 13%  372M 1s\n",
      "  1050K .......... .......... .......... .......... .......... 13%  166M 1s\n",
      "  1100K .......... .......... .......... .......... .......... 14% 8.69M 1s\n",
      "  1150K .......... .......... .......... .......... .......... 14%  269M 1s\n",
      "  1200K .......... .......... .......... .......... .......... 15% 22.0M 1s\n",
      "  1250K .......... .......... .......... .......... .......... 16% 2.57M 1s\n",
      "  1300K .......... .......... .......... .......... .......... 16% 69.2M 1s\n",
      "  1350K .......... .......... .......... .......... .......... 17% 4.57M 1s\n",
      "  1400K .......... .......... .......... .......... .......... 17% 65.4M 1s\n",
      "  1450K .......... .......... .......... .......... .......... 18%  180M 1s\n",
      "  1500K .......... .......... .......... .......... .......... 19% 5.49M 1s\n",
      "  1550K .......... .......... .......... .......... .......... 19%  114M 1s\n",
      "  1600K .......... .......... .......... .......... .......... 20% 7.88M 1s\n",
      "  1650K .......... .......... .......... .......... .......... 21% 6.59M 1s\n",
      "  1700K .......... .......... .......... .......... .......... 21% 73.7M 1s\n",
      "  1750K .......... .......... .......... .......... .......... 22% 14.9M 1s\n",
      "  1800K .......... .......... .......... .......... .......... 22% 4.31M 1s\n",
      "  1850K .......... .......... .......... .......... .......... 23% 1.87M 1s\n",
      "  1900K .......... .......... .......... .......... .......... 24% 92.4M 1s\n",
      "  1950K .......... .......... .......... .......... .......... 24% 49.0M 1s\n",
      "  2000K .......... .......... .......... .......... .......... 25% 13.5M 1s\n",
      "  2050K .......... .......... .......... .......... .......... 26% 6.24M 1s\n",
      "  2100K .......... .......... .......... .......... .......... 26% 67.6M 1s\n",
      "  2150K .......... .......... .......... .......... .......... 27% 79.1M 1s\n",
      "  2200K .......... .......... .......... .......... .......... 27% 4.86M 1s\n",
      "  2250K .......... .......... .......... .......... .......... 28% 94.8M 1s\n",
      "  2300K .......... .......... .......... .......... .......... 29% 4.48M 1s\n",
      "  2350K .......... .......... .......... .......... .......... 29% 7.86M 1s\n",
      "  2400K .......... .......... .......... .......... .......... 30% 27.3M 1s\n",
      "  2450K .......... .......... .......... .......... .......... 30% 3.10M 1s\n",
      "  2500K .......... .......... .......... .......... .......... 31% 64.7M 1s\n",
      "  2550K .......... .......... .......... .......... .......... 32% 82.8M 1s\n",
      "  2600K .......... .......... .......... .......... .......... 32% 10.8M 1s\n",
      "  2650K .......... .......... .......... .......... .......... 33% 90.0M 1s\n",
      "  2700K .......... .......... .......... .......... .......... 34% 5.29M 1s\n",
      "  2750K .......... .......... .......... .......... .......... 34% 56.3M 1s\n",
      "  2800K .......... .......... .......... .......... .......... 35% 5.53M 1s\n",
      "  2850K .......... .......... .......... .......... .......... 35%  135M 1s\n",
      "  2900K .......... .......... .......... .......... .......... 36% 3.52M 1s\n",
      "  2950K .......... .......... .......... .......... .......... 37% 34.8M 1s\n",
      "  3000K .......... .......... .......... .......... .......... 37% 9.28M 1s\n",
      "  3050K .......... .......... .......... .......... .......... 38%  155M 1s\n",
      "  3100K .......... .......... .......... .......... .......... 39% 4.57M 1s\n",
      "  3150K .......... .......... .......... .......... .......... 39% 57.5M 1s\n",
      "  3200K .......... .......... .......... .......... .......... 40%  182M 1s\n",
      "  3250K .......... .......... .......... .......... .......... 40% 3.73M 1s\n",
      "  3300K .......... .......... .......... .......... .......... 41% 83.8M 1s\n",
      "  3350K .......... .......... .......... .......... .......... 42%  191M 1s\n",
      "  3400K .......... .......... .......... .......... .......... 42% 3.88M 1s\n",
      "  3450K .......... .......... .......... .......... .......... 43% 40.2M 1s\n",
      "  3500K .......... .......... .......... .......... .......... 43% 5.15M 1s\n",
      "  3550K .......... .......... .......... .......... .......... 44% 48.2M 1s\n",
      "  3600K .......... .......... .......... .......... .......... 45%  146M 1s\n",
      "  3650K .......... .......... .......... .......... .......... 45% 3.83M 1s\n",
      "  3700K .......... .......... .......... .......... .......... 46%  103M 1s\n",
      "  3750K .......... .......... .......... .......... .......... 47%  152M 1s\n",
      "  3800K .......... .......... .......... .......... .......... 47%  544M 1s\n",
      "  3850K .......... .......... .......... .......... .......... 48% 5.68M 0s\n",
      "  3900K .......... .......... .......... .......... .......... 48%  232M 0s\n",
      "  3950K .......... .......... .......... .......... .......... 49% 2.19M 0s\n",
      "  4000K .......... .......... .......... .......... .......... 50% 8.45M 0s\n",
      "  4050K .......... .......... .......... .......... .......... 50% 45.0M 0s\n",
      "  4100K .......... .......... .......... .......... .......... 51% 4.58M 0s\n",
      "  4150K .......... .......... .......... .......... .......... 52%  117M 0s\n",
      "  4200K .......... .......... .......... .......... .......... 52% 19.5M 0s\n",
      "  4250K .......... .......... .......... .......... .......... 53%  102M 0s\n",
      "  4300K .......... .......... .......... .......... .......... 53% 2.69M 0s\n",
      "  4350K .......... .......... .......... .......... .......... 54% 83.6M 0s\n",
      "  4400K .......... .......... .......... .......... .......... 55%  121M 0s\n",
      "  4450K .......... .......... .......... .......... .......... 55% 9.85M 0s\n",
      "  4500K .......... .......... .......... .......... .......... 56%  102M 0s\n",
      "  4550K .......... .......... .......... .......... .......... 57%  261M 0s\n",
      "  4600K .......... .......... .......... .......... .......... 57% 1.84M 0s\n",
      "  4650K .......... .......... .......... .......... .......... 58% 6.32M 0s\n",
      "  4700K .......... .......... .......... .......... .......... 58% 49.2M 0s\n",
      "  4750K .......... .......... .......... .......... .......... 59% 10.8M 0s\n",
      "  4800K .......... .......... .......... .......... .......... 60% 5.01M 0s\n",
      "  4850K .......... .......... .......... .......... .......... 60%  271M 0s\n",
      "  4900K .......... .......... .......... .......... .......... 61%  115M 0s\n",
      "  4950K .......... .......... .......... .......... .......... 61% 5.14M 0s\n",
      "  5000K .......... .......... .......... .......... .......... 62% 50.3M 0s\n",
      "  5050K .......... .......... .......... .......... .......... 63% 3.50M 0s\n",
      "  5100K .......... .......... .......... .......... .......... 63%  160M 0s\n",
      "  5150K .......... .......... .......... .......... .......... 64% 15.1M 0s\n",
      "  5200K .......... .......... .......... .......... .......... 65%  306M 0s\n",
      "  5250K .......... .......... .......... .......... .......... 65%  202M 0s\n",
      "  5300K .......... .......... .......... .......... .......... 66%  164M 0s\n",
      "  5350K .......... .......... .......... .......... .......... 66% 7.69M 0s\n",
      "  5400K .......... .......... .......... .......... .......... 67% 8.07M 0s\n",
      "  5450K .......... .......... .......... .......... .......... 68% 75.0M 0s\n",
      "  5500K .......... .......... .......... .......... .......... 68% 5.82M 0s\n",
      "  5550K .......... .......... .......... .......... .......... 69% 4.58M 0s\n",
      "  5600K .......... .......... .......... .......... .......... 70% 6.70M 0s\n",
      "  5650K .......... .......... .......... .......... .......... 70% 34.4M 0s\n",
      "  5700K .......... .......... .......... .......... .......... 71%  281M 0s\n",
      "  5750K .......... .......... .......... .......... .......... 71% 11.8M 0s\n",
      "  5800K .......... .......... .......... .......... .......... 72% 65.4M 0s\n",
      "  5850K .......... .......... .......... .......... .......... 73% 54.6M 0s\n",
      "  5900K .......... .......... .......... .......... .......... 73% 2.49M 0s\n",
      "  5950K .......... .......... .......... .......... .......... 74% 94.0M 0s\n",
      "  6000K .......... .......... .......... .......... .......... 74%  307M 0s\n",
      "  6050K .......... .......... .......... .......... .......... 75%  263M 0s\n",
      "  6100K .......... .......... .......... .......... .......... 76%  288M 0s\n",
      "  6150K .......... .......... .......... .......... .......... 76% 8.37M 0s\n",
      "  6200K .......... .......... .......... .......... .......... 77% 3.78M 0s\n",
      "  6250K .......... .......... .......... .......... .......... 78% 98.7M 0s\n",
      "  6300K .......... .......... .......... .......... .......... 78% 2.62M 0s\n",
      "  6350K .......... .......... .......... .......... .......... 79%  157M 0s\n",
      "  6400K .......... .......... .......... .......... .......... 79%  424M 0s\n",
      "  6450K .......... .......... .......... .......... .......... 80% 3.23M 0s\n",
      "  6500K .......... .......... .......... .......... .......... 81% 30.9M 0s\n",
      "  6550K .......... .......... .......... .......... .......... 81%  452M 0s\n",
      "  6600K .......... .......... .......... .......... .......... 82% 8.21M 0s\n",
      "  6650K .......... .......... .......... .......... .......... 83% 5.23M 0s\n",
      "  6700K .......... .......... .......... .......... .......... 83% 9.57M 0s\n",
      "  6750K .......... .......... .......... .......... .......... 84% 3.61M 0s\n",
      "  6800K .......... .......... .......... .......... .......... 84% 93.1M 0s\n",
      "  6850K .......... .......... .......... .......... .......... 85% 4.97M 0s\n",
      "  6900K .......... .......... .......... .......... .......... 86% 41.2M 0s\n",
      "  6950K .......... .......... .......... .......... .......... 86%  494M 0s\n",
      "  7000K .......... .......... .......... .......... .......... 87% 5.51M 0s\n",
      "  7050K .......... .......... .......... .......... .......... 87%  158M 0s\n",
      "  7100K .......... .......... .......... .......... .......... 88% 5.97M 0s\n",
      "  7150K .......... .......... .......... .......... .......... 89% 79.3M 0s\n",
      "  7200K .......... .......... .......... .......... .......... 89% 65.0M 0s\n",
      "  7250K .......... .......... .......... .......... .......... 90% 4.07M 0s\n",
      "  7300K .......... .......... .......... .......... .......... 91% 89.6M 0s\n",
      "  7350K .......... .......... .......... .......... .......... 91%  149M 0s\n",
      "  7400K .......... .......... .......... .......... .......... 92% 10.1M 0s\n",
      "  7450K .......... .......... .......... .......... .......... 92% 73.1M 0s\n",
      "  7500K .......... .......... .......... .......... .......... 93% 51.8M 0s\n",
      "  7550K .......... .......... .......... .......... .......... 94% 15.4M 0s\n",
      "  7600K .......... .......... .......... .......... .......... 94% 2.93M 0s\n",
      "  7650K .......... .......... .......... .......... .......... 95%  101M 0s\n",
      "  7700K .......... .......... .......... .......... .......... 96%  120M 0s\n",
      "  7750K .......... .......... .......... .......... .......... 96%  133M 0s\n",
      "  7800K .......... .......... .......... .......... .......... 97% 49.0M 0s\n",
      "  7850K .......... .......... .......... .......... .......... 97%  314M 0s\n",
      "  7900K .......... .......... .......... .......... .......... 98%  117M 0s\n",
      "  7950K .......... .......... .......... .......... .......... 99% 9.48M 0s\n",
      "  8000K .......... .......... .......... .......... .......... 99% 2.76M 0s\n",
      "  8050K .......... ........                                   100%  223M=0.9s\n",
      "\n",
      "2025-03-07 19:27:08 (9.10 MB/s) - 'green_tripdata_2019-10.csv.gz' saved [8262584/8262584]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://github.com/DataTalksClub/nyc-tlc-data/releases/download/green/green_tripdata_2019-10.csv.gz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "57fb14bf-f7f2-45a9-b918-d64203e5d802",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "2b8b3ac1-e3fb-4713-9ccb-7c0fbfe4c017",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\Users\\alexe\\AppData\\Local\\Temp\\ipykernel_3424\\2667354967.py:1: DtypeWarning: Columns (3) have mixed types. Specify dtype option on import or set low_memory=False.\n",
      "  df = pd.read_csv('green_tripdata_2019-10.csv.gz')\n"
     ]
    }
   ],
   "source": [
    "df = pd.read_csv('green_tripdata_2019-10.csv.gz')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a0e8ab41-1520-46b1-b8fa-a3fedf170896",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>VendorID</th>\n",
       "      <th>lpep_pickup_datetime</th>\n",
       "      <th>lpep_dropoff_datetime</th>\n",
       "      <th>store_and_fwd_flag</th>\n",
       "      <th>RatecodeID</th>\n",
       "      <th>PULocationID</th>\n",
       "      <th>DOLocationID</th>\n",
       "      <th>passenger_count</th>\n",
       "      <th>trip_distance</th>\n",
       "      <th>fare_amount</th>\n",
       "      <th>extra</th>\n",
       "      <th>mta_tax</th>\n",
       "      <th>tip_amount</th>\n",
       "      <th>tolls_amount</th>\n",
       "      <th>ehail_fee</th>\n",
       "      <th>improvement_surcharge</th>\n",
       "      <th>total_amount</th>\n",
       "      <th>payment_type</th>\n",
       "      <th>trip_type</th>\n",
       "      <th>congestion_surcharge</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2.0</td>\n",
       "      <td>2019-10-01 00:26:02</td>\n",
       "      <td>2019-10-01 00:39:58</td>\n",
       "      <td>N</td>\n",
       "      <td>1.0</td>\n",
       "      <td>112</td>\n",
       "      <td>196</td>\n",
       "      <td>1.0</td>\n",
       "      <td>5.88</td>\n",
       "      <td>18.0</td>\n",
       "      <td>0.50</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.3</td>\n",
       "      <td>19.30</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1.0</td>\n",
       "      <td>2019-10-01 00:18:11</td>\n",
       "      <td>2019-10-01 00:22:38</td>\n",
       "      <td>N</td>\n",
       "      <td>1.0</td>\n",
       "      <td>43</td>\n",
       "      <td>263</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.80</td>\n",
       "      <td>5.0</td>\n",
       "      <td>3.25</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.3</td>\n",
       "      <td>9.05</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1.0</td>\n",
       "      <td>2019-10-01 00:09:31</td>\n",
       "      <td>2019-10-01 00:24:47</td>\n",
       "      <td>N</td>\n",
       "      <td>1.0</td>\n",
       "      <td>255</td>\n",
       "      <td>228</td>\n",
       "      <td>2.0</td>\n",
       "      <td>7.50</td>\n",
       "      <td>21.5</td>\n",
       "      <td>0.50</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.3</td>\n",
       "      <td>22.80</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1.0</td>\n",
       "      <td>2019-10-01 00:37:40</td>\n",
       "      <td>2019-10-01 00:41:49</td>\n",
       "      <td>N</td>\n",
       "      <td>1.0</td>\n",
       "      <td>181</td>\n",
       "      <td>181</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.90</td>\n",
       "      <td>5.5</td>\n",
       "      <td>0.50</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.3</td>\n",
       "      <td>6.80</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2.0</td>\n",
       "      <td>2019-10-01 00:08:13</td>\n",
       "      <td>2019-10-01 00:17:56</td>\n",
       "      <td>N</td>\n",
       "      <td>1.0</td>\n",
       "      <td>97</td>\n",
       "      <td>188</td>\n",
       "      <td>1.0</td>\n",
       "      <td>2.52</td>\n",
       "      <td>10.0</td>\n",
       "      <td>0.50</td>\n",
       "      <td>0.5</td>\n",
       "      <td>2.26</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.3</td>\n",
       "      <td>13.56</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   VendorID lpep_pickup_datetime lpep_dropoff_datetime store_and_fwd_flag  \\\n",
       "0       2.0  2019-10-01 00:26:02   2019-10-01 00:39:58                  N   \n",
       "1       1.0  2019-10-01 00:18:11   2019-10-01 00:22:38                  N   \n",
       "2       1.0  2019-10-01 00:09:31   2019-10-01 00:24:47                  N   \n",
       "3       1.0  2019-10-01 00:37:40   2019-10-01 00:41:49                  N   \n",
       "4       2.0  2019-10-01 00:08:13   2019-10-01 00:17:56                  N   \n",
       "\n",
       "   RatecodeID  PULocationID  DOLocationID  passenger_count  trip_distance  \\\n",
       "0         1.0           112           196              1.0           5.88   \n",
       "1         1.0            43           263              1.0           0.80   \n",
       "2         1.0           255           228              2.0           7.50   \n",
       "3         1.0           181           181              1.0           0.90   \n",
       "4         1.0            97           188              1.0           2.52   \n",
       "\n",
       "   fare_amount  extra  mta_tax  tip_amount  tolls_amount  ehail_fee  \\\n",
       "0         18.0   0.50      0.5        0.00           0.0        NaN   \n",
       "1          5.0   3.25      0.5        0.00           0.0        NaN   \n",
       "2         21.5   0.50      0.5        0.00           0.0        NaN   \n",
       "3          5.5   0.50      0.5        0.00           0.0        NaN   \n",
       "4         10.0   0.50      0.5        2.26           0.0        NaN   \n",
       "\n",
       "   improvement_surcharge  total_amount  payment_type  trip_type  \\\n",
       "0                    0.3         19.30           2.0        1.0   \n",
       "1                    0.3          9.05           2.0        1.0   \n",
       "2                    0.3         22.80           2.0        1.0   \n",
       "3                    0.3          6.80           2.0        1.0   \n",
       "4                    0.3         13.56           1.0        1.0   \n",
       "\n",
       "   congestion_surcharge  \n",
       "0                   0.0  \n",
       "1                   0.0  \n",
       "2                   0.0  \n",
       "3                   0.0  \n",
       "4                   0.0  "
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "d085b583-1609-41a9-a222-ff6ca495ee27",
   "metadata": {},
   "outputs": [],
   "source": [
    "columns = [\n",
    "    'lpep_pickup_datetime',\n",
    "    'lpep_dropoff_datetime',\n",
    "    'PULocationID',\n",
    "    'DOLocationID',\n",
    "    'passenger_count',\n",
    "    'trip_distance',\n",
    "    'tip_amount'\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "66e9f47c-9284-4760-8011-3a8f48aaa49f",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df[columns]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "7ae3f843-d428-43d2-9e47-7f9fb43acbad",
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "f1ca1ac1-176c-4ccc-aa11-7e1cb5659d39",
   "metadata": {},
   "outputs": [],
   "source": [
    "from tqdm.auto import tqdm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "0b3da4e1-2f1c-400f-bb67-82734c1193f4",
   "metadata": {},
   "outputs": [],
   "source": [
    "messages = df.to_dict(orient='records')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "3bdc95d8-64e1-4819-a885-996813b4bf94",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "476386"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(messages)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "d6f15929-e928-464d-afc1-690343f4f780",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "4dffdeb2a0064e1d9bd02dff9f9c49f0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "  0%|          | 0/476386 [00:00<?, ?it/s]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "topic_name = 'green-trips'\n",
    "\n",
    "for message in tqdm(messages):\n",
    "    producer.send(topic_name, value=message)\n",
    "\n",
    "producer.flush()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2409953c-e9dd-403d-a0d1-d8b883c23ef5",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
